package com.apes.fn.misc.taskDownload;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.apes.framework.api.DomainService;
import com.apes.framework.api.SpringManager;
import com.apes.framework.jpa.Connector;
import com.apes.framework.jpa.spec.JpaDsl;
import com.apes.framework.plugin.esb.api.SimpleRequest;
import com.apes.framework.plugin.ftp.FtpUtil;
import com.apes.framework.plugin.websocket.WsMessage;
import com.apes.framework.plugin.websocket.WsPublisher;
import com.apes.framework.query.QueryService;
import com.apes.framework.query.lib.PubToolUtil;
import com.apes.framework.util.DistributedLock;
import com.apes.framework.util.MapUtil;
import com.apes.oap.Constants;
import com.apes.scm.rbac.model.User;
import com.apestech.framework.db.SqlResultSet;
import com.apestech.framework.util.DateUtil;
import org.apache.commons.lang.StringUtils;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionTemplate;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.FileSystems;
import java.security.GeneralSecurityException;
import java.security.MessageDigest;
import java.sql.SQLException;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import com.apestech.framework.db.SqlSession;

/**
 * 任务下载类
 *
 * @author wuq
 * @Time 2023-6-13 10:08
 * @Description
 */
@Service
public class TaskDownloadService extends DomainService {

    @Autowired
    private TaskDownloadRepository taskDownloadRepository;
    @Autowired
    private Connector connector;
    @Autowired
    private FtpUtil ftpUtil;
    @Autowired
    private WsPublisher wsPublisher;
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private DistributedLock distributedLock;


    /**
     * 创建下载任务
     *
     * @param request {"method":"调用方法/查询API", "body":"传入参数"}
     */
    public TaskDownload create(SimpleRequest request) throws IOException {
        JSONObject json = request.getJO();

        // 校验是否已经创建任务, 如果存在已经下载的任务就返回, 重复的策略：同一天，同一个界面，相同的查询条件
        String md5 = encodeParams(json.getJSONArray("condition"), json.getString("queryid"), request.getPostId());
        List<TaskDownload> taskDownloads = taskDownloadRepository.findAllByMd5(md5);
        if (taskDownloads.size() > 0) return taskDownloads.get(0);

        TaskDownload taskDownload = new TaskDownload();
        taskDownload.setApi(json.getString("queryid"));
        taskDownload.setParams(json.toString());
        taskDownload.setName(json.getString("queryname"));
        taskDownload.setState("create");
        taskDownload.setMd5(md5);

        // 下载人相关信息
        taskDownload.setPostId(request.getPostId());
        taskDownload.setCreator(request.getPersonId());
        taskDownload.setUserId(((User) request.getCurrentUser()).getId());
        taskDownload = taskDownloadRepository.saveAndFlush(taskDownload);
        TaskDownload finalTaskDownload = taskDownload;
        CompletableFuture.runAsync(() -> execute(() -> coreProcess(finalTaskDownload)));
        return taskDownload;
    }

    @Autowired
    private TransactionTemplate transactionTemplate;

    public void execute(Callback callback) {
        transactionTemplate.execute(status -> {
            try {
                callback.execute();
                return null;
            } catch (Exception e) {
                status.isRollbackOnly();
                throw e;
            }
        });
    }

    @FunctionalInterface
    public interface Callback {
        @Nullable
        void execute();
    }

    /**
     * 补充定时任务下载的逻辑
     */
    public void schedule() {
        // 查询对应的下载任务
        List<TaskDownload> taskDownloads = taskDownloadRepository.findForFile();

        taskDownloads.forEach(taskDownload -> {
            String lockId = getLockId(taskDownload.getId());
            if (distributedLock.existLock(lockId)) return;
            saveLog(taskDownload.getId(), "开始执行");
            try {
                execute(() -> coreProcess(taskDownload));
            } catch (Exception e) {
                saveScheduleError(taskDownload, e.getMessage());
            } finally {
                saveLog(taskDownload.getId(), "执行结束");
            }
        });
    }

    private void saveLog(String taskId, String remark) {
        Document document = new Document();
        document.put("taskId", taskId);
        document.put("remark", remark);
        document.put("date", LocalDateTime.now());
        mongoTemplate.getCollection("taskDownLoad").insertOne(document);
    }

    private void saveScheduleError(TaskDownload taskDownload, String error){
        // 单独的事务处理保存操作
        execute( () -> {
            TaskDownload task = taskDownloadRepository.findOne(taskDownload.getId());
            task.setSuccess(true);
            task.setError(error);
            taskDownloadRepository.saveAndFlush(task);
        });
    }



    public void test(SimpleRequest request) throws Exception {
        String id = request.get("id");
        TaskDownload taskDownload = taskDownloadRepository.findOne(id);

        // 核心处理方法
        coreProcess(taskDownload);
    }

    /**
     * 核心处理方法
     * <p>
     * 查询SQL 数据，传入到 FTP 中
     * </p>
     *
     * @param taskDownload TaskDownload
     */
    private void coreProcess(TaskDownload taskDownload) {
        invoke("task.download.process", MapUtil.mapper("taskDownloadId", taskDownload.getId()));
    }

    public Map doProcess(SimpleRequest request) throws Exception {
        AtomicBoolean isDone = new AtomicBoolean(false);
        String taskDownloadId = request.get("taskDownloadId");
        String lockId = getLockId(taskDownloadId);
        boolean isLocked = false;
        try {
            isLocked = distributedLock.lock(lockId, 2 * 60 * 60 * 1000);
            if (!isLocked) return MapUtil.mapper("taskDownloadId", taskDownloadId, "isDone", isDone.get());

            TaskDownload taskDownload = taskDownloadRepository.findOne(JpaDsl.toCriteriaByEq("id", taskDownloadId, "state", "create")).orElse(null);
            if (taskDownload == null) return MapUtil.mapper("taskDownloadId", taskDownloadId, "isDone", isDone.get());;

            // 传入FTP上面的文件名
            String fileName = taskDownload.getName() + "_" + DateUtil.dateTimeToStr(taskDownload.getCreateDate()) + ".xlsx";

            com.apestech.framework.json.JSONObject params = new com.apestech.framework.json.JSONObject(taskDownload.getParams());
            PubToolUtil.getRole(params, taskDownload.getPostId());
            params.put("postId", taskDownload.getPostId());

            // 获取头部数据
            List<Map> gridHeaders = params.getObject("gridheader", List.class);

            // 处理查询条件
            com.apestech.framework.json.JSONObject conditionJson;
            if (params.getBoolean("customAction")) {
                conditionJson = invoke(params.getString("queryid"), params);
            } else {
                conditionJson = params;
            }

            // 查询数据
            SqlSession sqlSession = PubToolUtil.getSession(null, connector.getConnection());
            sqlSession.getForResultSet().queryId(params.getString("queryid"), conditionJson, (rs, rsList) -> {
                /* 获取数据库返回值列，避免 xml 配置不存在的列报错 */
                ArrayList<Object> arrayList = rs.toHeadArray();     //  获取数据数据库返回的数据
                // 传入到 FTP 中
                boolean done = transferFtp(gridHeaders, rs, arrayList, fileName);
                isDone.set(done);
            });

           return MapUtil.mapper("taskDownloadId", taskDownloadId, "fileName", fileName, "isDone", isDone.get());
        } finally {
            if (isLocked) distributedLock.unlock(lockId);
        }
    }

    private String getLockId(String taskDownloadId) {
        return String.format("%s-%s", "taskDownload", taskDownloadId);
    }

    public void done(SimpleRequest request) {
        boolean isDone = request.get("isDone");
        if (!isDone) return;
        String fileName = request.get("fileName");
        String taskDownloadId = request.get("taskDownloadId");
        TaskDownload taskDownload = taskDownloadRepository.findOne(JpaDsl.toCriteriaByEq("id", taskDownloadId, "state", "create")).orElse(null);
        if (taskDownload == null) return;
        // 回写下载路径到下载任务表中
        taskDownload.setFtpName(fileName);
        taskDownload.setState("finish");    // 修改状态为已完成
        taskDownload.setSuccess(true);
        taskDownload.setError(null);
        TaskDownload td = taskDownloadRepository.saveAndFlush(taskDownload);
        sendFinishMessage(td);  // 发送完成消息
    }

    private boolean transferFtp(List<Map> gridHeaders, SqlResultSet rs, ArrayList<Object> arrayList, String fileName) throws Exception {
        // 过滤导出数据列
        List filterHeaders = gridHeaders.stream().filter(gridHeader -> {
            Map m = (Map) gridHeader;
            return arrayList.contains(m.get("field").toString());
        }).collect(Collectors.toList());

        List<List<String>> excelHeaders = getExcelHeaders(filterHeaders);  // 获取导出 Excel 头部
        List<String> columnName = getColumnName(filterHeaders);    // 获取每一行的数据列名

        // 传入数据到 ftp 中
        boolean isDone = ftpUtil.uploadExcel(excelHeaders, fileName, rs, resultSet -> {
            List row = new ArrayList();
            columnName.forEach(v -> {
                try {
                    // 对字段值做处理
                    row.add(formateFieldValue(resultSet.getObject(v)));
                } catch (SQLException e) {
                    throw new RuntimeException(e.getMessage());
                }
            });
            return row;
        });
        return isDone;
    }


    /**
     * 对 ResultSet 中的返回值进行格式化
     *
     * @param fieldValue field
     * @return String
     * @throws SQLException
     */
    private Object formateFieldValue(Object fieldValue) throws SQLException {
        if (fieldValue == null) {
            fieldValue = "";
        } else {
            if (fieldValue instanceof oracle.sql.TIMESTAMP) {
                oracle.sql.TIMESTAMP time = (oracle.sql.TIMESTAMP) fieldValue;
                fieldValue = DateUtil.dateTimeToStr(new Date(time.timestampValue().getTime()));
            }
            if (fieldValue instanceof java.sql.Timestamp) {
                java.sql.Timestamp time = (java.sql.Timestamp) fieldValue;
                fieldValue = DateUtil.dateTimeToStr(new Date(time.getTime()));
            }
        }
        return fieldValue;
    }

    private void sendFinishMessage(TaskDownload taskDownload) {
        WsMessage wsMessage = new WsMessage();
        wsMessage.setTo(taskDownload.getUserId().toString());

        JSONObject json = new JSONObject();
        json.put("topic", "ws.taskDownload");
        json.put("id", taskDownload.getId());
        json.put("downloadUrl", taskDownload.getDownloadUrl());
        wsMessage.setBody(json.toJSONString());
        wsPublisher.publish(wsMessage);
    }


    /**
     * 获取 Excel 头部
     *
     * @param gridHeader 类似 [{"field":"ID", "title":"零售订单号"}]
     * @return List<List < String>> 类似 [[公司编码], [公司名称]]
     */
    private List<List<String>> getExcelHeaders(List<Map> gridHeader) {
        List<List<String>> head = new ArrayList<>();
        gridHeader.forEach(v -> head.add(Collections.singletonList(v.get("title").toString())));
        return head;
    }

    /**
     * 获取 Excel 头部
     *
     * @param gridHeader 类似 [{"field":"ID", "title":"零售订单号"}]
     * @return List < String> 类似 [ "", ""]
     */
    private List<String> getColumnName(List<Map> gridHeader) {
        return gridHeader.stream().map(v -> v.get("field").toString()).collect(Collectors.toList());
    }


    /**
     * 标记已下载，用于后续清理下载完的文件
     *
     * @param request {"id":""}
     */
    public void downloaded(SimpleRequest request) {
        String id = request.get("id");
        TaskDownload taskDownload = taskDownloadRepository.findOne(id);
        taskDownload.setState("download");
        taskDownloadRepository.saveAndFlush(taskDownload);
    }

    /**
     * 定时任务删除文件, 删除指定日期之前的文件
     *
     * @param request {"daysAgo": 4}
     */
    public void remove(SimpleRequest request) {
        JSONObject jsonObject = request.getJO();
        int daysAgo = jsonObject.getInteger("daysAgo");

        Date date = com.apes.framework.util.DateUtil.add(new Date(), Calendar.DATE, -1 * daysAgo);

        // 链接到FTP 删除文件
       List<TaskDownload> taskDownloads = taskDownloadRepository.findAllForDelete(date);
       taskDownloads.forEach(taskDownload -> {
           // 每个都是独立的事务处理
           execute( () -> {
               try {
                   // 第一步：删除ftp文件
                   boolean flag = ftpUtil.removeFile(taskDownload.getFtpName());
                   if (!flag) throw new RuntimeException("删除FTP文件失败");

                   // 第二步：删除表中的数据
                   taskDownloadRepository.delete(taskDownload);
               } catch (Exception e) {
                   throw new RuntimeException(e.getMessage());
               }
           });
       });
    }

    /**
     * 测试删除任务
     * @param request {"id":""}
     */
    public void testRemove(SimpleRequest request){
        String id = request.get("id");
        TaskDownload taskDownload = taskDownloadRepository.findOne(id);

        try {
            // 第一步：删除ftp文件
            boolean flag = ftpUtil.removeFile(taskDownload.getFtpName());
            if (!flag) throw new RuntimeException("删除FTP文件失败");

            // 第二步：删除表中的数据
            taskDownloadRepository.delete(taskDownload);
        } catch (Exception e) {
           throw new RuntimeException(e.getMessage());
        }
    }

    /**
     * 对参数进行 MD5 加密
     *
     * @param params  界面请求参数
     * @param queryId 界面查询ID
     * @return md5
     */
    private String encodeParams(JSONArray params, String queryId, String postId) throws IOException {
        byte[] sha1Digest = getSHA1Digest(queryId + params.toString() + postId + LocalDate.now());
        return byte2hex(sha1Digest);
    }

    private byte[] getSHA1Digest(String data) throws IOException {
        byte[] bytes = null;
        try {
            MessageDigest md = MessageDigest.getInstance("SHA-1");
            bytes = md.digest(data.getBytes(Constants.UTF8));
        } catch (GeneralSecurityException gse) {
            throw new IOException(gse.getMessage());
        }
        return bytes;
    }

    /**
     * 二进制转十六进制字符串
     *
     * @param bytes
     * @return
     */
    private String byte2hex(byte[] bytes) {
        StringBuilder sign = new StringBuilder();
        for (int i = 0; i < bytes.length; i++) {
            String hex = Integer.toHexString(bytes[i] & 0xFF);
            if (hex.length() == 1) {
                sign.append("0");
            }
            sign.append(hex.toUpperCase());
        }
        return sign.toString();
    }

}
